package com.sailing.lianxi.kafka;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;

import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.admin.TopicCommand;
import kafka.admin.TopicCommand.TopicCommandOptions;
import kafka.server.ConfigType;
import kafka.utils.ZkUtils;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.security.JaasUtils;

import com.sailing.lianxi.common.Constants;

public class KAFKASupport {

	//超时时间毫秒数
	static int OUT_TIME=1000;
	
	/**
	 * 测试kafka是否可用
	 * @param url
	 * @return
	 */
	public static boolean testKafkaConn(String kafkaServerUrl){
		boolean flag = false;
		 KafkaConsumer<byte[], byte[]> consumer = null;
		try {
		    Properties props = new Properties();
	        props.put("bootstrap.servers",kafkaServerUrl);// 该地址是集群的子集，用来探测集群。
	        props.put("key.deserializer",
	                "org.apache.kafka.common.serialization.ByteArrayDeserializer");// 反序列化器
	        props.put("value.deserializer",
	                "org.apache.kafka.common.serialization.ByteArrayDeserializer");
	        props.put("session.timeout.ms", "60000");
	        props.put("request.timeout.ms", "65000");
	        props.put("consumer.id",UUID.randomUUID().toString());
	        consumer = new KafkaConsumer<byte[], byte[]>(props);
	        consumer.listTopics();
			flag =true;
		} catch (Exception e) {
		System.out.println("测试连接失败");
		e.printStackTrace();
		}finally{
			if(null!=consumer){
				consumer.close();
			}
		}
		return flag;
	}
	
	/**
	 * 创建topic
	 * @param topicName
	 * @return
	 */
	public static boolean createTopic(String topicName){
		ZkUtils zkUtils = null;
		try {
			zkUtils = getZkUtils();
			AdminUtils.createTopic(zkUtils, topicName, Constants.KAFKA_PARTITION_NUM,Constants.KAFKA__REPLICA_NUM , new Properties(), RackAwareMode.Enforced$.MODULE$);
			System.out.println("创建topic:"+topicName+" 成功");
			return true;
		} catch (Exception e) {
			System.out.println("创建topic:"+topicName+" 失败");
			e.printStackTrace();
			return false;
		}finally{
			closeZkUtils(zkUtils);
		}
	}
	
	/**
	 * 删除topic
	 * @param topicName topic名称
	 * @注意事项：
	 * 1)属性delete.topic.enable必须等于true，该方法才起作用
	 * 2) topic不被使用，如果正在使用，执行该方法也不会报错，只是对topic执行逻辑删除
	 * 3)删除比较费时，可能你还能看到删除过程产生的文件-delete
	 * @return
	 */
	public static boolean deleteTopic(String topicName){
		ZkUtils zkUtils = null;
		try {
			boolean flag = isExistTopic(topicName);
			if(flag){
				zkUtils = getZkUtils();
				AdminUtils.deleteTopic(zkUtils, topicName);
				System.out.println("删除:"+topicName+"完成......");
				return true;
//				//这里的内部topic是：__consumer_offsets
//				if(!Topic.isInternal(topicName)){
//					AdminUtils.deleteTopic(zkUtils, topicName);
//					System.out.println("删除:"+topicName+"完成......");
//					return true;
//				}else{
//					System.out.println("内部topic:"+topicName+"不能被删除");
//					return false;
//				}
			}else{
				System.out.println("要删除的topic不存在");
				return false;
			}
		
		} catch (Exception e) {
			System.out.println("删除topic失败");
			e.printStackTrace();
			return false;
		}finally{
			closeZkUtils(zkUtils);
		}
	}
	
	/**
	 * topic是否存在
	 * @param topicName
	 * @return
	 */
	public static boolean isExistTopic(String topicName){
		ZkUtils zkUtils = null;
		try {
			zkUtils = getZkUtils();
			return AdminUtils.topicExists(zkUtils, topicName);
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		}finally{
			closeZkUtils(zkUtils);
		}
	}
	
	/**
	 * 查询所有topic名称
	 * @return
	 */
	public static  String[] getAllTopic(){
		ZkUtils zkUtils = null;
		try {
			zkUtils = getZkUtils();
			if(zkUtils.getAllTopics().size()>0){
				return zkUtils.getAllTopics().mkString(",").split(",");
			}else{
				return new String[0];
			}
		} catch (Exception e) {
			System.out.println("查询所有topic失败");
			e.printStackTrace();
			return null;
		}finally{
			closeZkUtils(zkUtils);
		}
	}
	
	
	/**
	 * 获取topic信息
	 * @param topicName
	 * @return
	 */
	public static Map<String, Object> getTopicInfo(String topicName){
		ZkUtils zkUtils = null;
		Map<String,Object> map = null;
		try {
			map = new HashMap<String, Object>();
			zkUtils = getZkUtils();
		
			Properties props = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic(), topicName);
			// 增加topic级别属性
			props.put("min.cleanable.dirty.ratio", "0.3");
			// 删除topic级别属性
			props.remove("max.message.bytes");
			// 修改topic 'test'的属性
			AdminUtils.changeTopicConfig(zkUtils,topicName, props);
			Iterator it = props.entrySet().iterator();
			while(it.hasNext()){
				Map.Entry entry = (Map.Entry)it.next();
				map.put(entry.getKey().toString(), entry.getValue());
			}
		} catch (Exception e) {
			e.printStackTrace();
		}finally{
			closeZkUtils(zkUtils);
		}
		return map;
	}
	
	/**
	 * 修改topicName属性
	 * @param topicName
	 * @return
	 */
	public static boolean updateTopicConf(String topicName){
		ZkUtils zkUtils = null;
		try {
			zkUtils = getZkUtils();
			Properties props = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic(), topicName);
			// 增加topic级别属性
			props.put("min.cleanable.dirty.ratio", "0.3");
			// 删除topic级别属性
			props.remove("max.message.bytes");
			// 修改topic 'test'的属性
			AdminUtils.changeTopicConfig(zkUtils, topicName, props);
			return true;
		} catch (Exception e) {
			System.out.println("修改topic属性失败");
			e.printStackTrace();
			return false;
		}finally{
			closeZkUtils(zkUtils);
		}
	}
	
   
	/**
	 * 获取topic分区信息
	 * @param topicName topic名称
	 */
	public static void getPartition(String topicName){
//		ZkUtils zkUtils = getZkUtils();
//	     String[] opts = new String[]{
//	             "--topic",
//	             topicName, 
//	               "--describe"   
//	     };
//	     TopicCommandOptions cmd = new TopicCommandOptions(opts);
//	     TopicCommand.describeTopic(zkUtils, cmd);
	}
	
	/**
	 * 动态增加分区
	 * @param topicName
	 * @return
	 */
	public static boolean addPartition(String topicName){
		ZkUtils zkUtils = getZkUtils();
//		AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkUtils, topicName,  AdminUtils.assignReplicasToBrokers(AdminUtils.getBrokerMetadatas(zkUtils, RackAwareMode.Enforced$.MODULE$, zkUtils.getSortedBrokerList()), 4, 1, -1, -1), new Properties(),true);
		return true;
	}
	
	/**
	 * 获取ZkUtils对象
	 * @return
	 */
	public static ZkUtils getZkUtils(){
		ZkUtils zkUtils = null;
		try {
			zkUtils = ZkUtils.apply(Constants.KAFKA_ZOOKEEPER_URL, OUT_TIME, OUT_TIME, JaasUtils.isZkSecurityEnabled());
		} catch (Exception e) {
			System.out.println("获取ZkUtils对象失败");
		   e.printStackTrace();
		}
		return zkUtils;
	}
	
	/**
	 * 关闭zkUtils
	 * @param zkUtils
	 */
	private static void closeZkUtils(ZkUtils zkUtils){
		try {
			if(null!= zkUtils){
				zkUtils.close();
			}
		} catch (Exception e) {
		  System.out.println("关闭zkUtils报错");
		  e.printStackTrace();
		}
		
	}
	
	
	
	public static void main(String[] args) {
		String topicName= Constants.TOPIC_NAME;
//		System.out.println(testKafkaConn(Constants.KAFKA_SERVERS));
		createTopic(topicName);
//		String topicName = Constants.TOPIC_NAME;
//		deleteTopic(topicName);
//		createTopic(topicName);
//		
		
//		String [] allTopics = getAllTopic(); 
//	     for (String topicName : allTopics) {
//	         System.out.println("topicName="+topicName);
////	         if(topicName.lastIndexOf("wg_")!=-1){
//	             deleteTopic(topicName);    
////	         }
//		}
	}
	
}
